package com.syygl.test.middleware.consumer;

import com.syygl.test.middleware.core.rocketMQ.ConsumerProperties;
import com.syygl.test.middleware.core.rocketMQ.DefaultProperties;
import com.syygl.test.middleware.utils.SpringContextUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

public class InitConsumerInstance {

    private DefaultProperties defaultProperties;
    private List<ConsumerProperties> consumerPropertiesList;


    public InitConsumerInstance(DefaultProperties defaultProperties, List<ConsumerProperties> consumerPropertiesList) {
        this.defaultProperties = defaultProperties;
        this.consumerPropertiesList = consumerPropertiesList;


        initConcurrentlyConsumer();


    }

    private void initConcurrentlyConsumer() {
        this.consumerPropertiesList.forEach(cp -> {
            ConsumeMsgConcurrently consumeMsgConcurrently =
                    (ConsumeMsgConcurrently) SpringContextUtils.getBean(cp.getConsumerMessageConcurrently());

            Map<String, Object> pros = consumeMsgConcurrently.consumerObjectProperties();
            DefaultMQPushConsumer consumer = getConsumerInstance(cp, pros);

            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    //consumeMsgConcurrently.



                   // String s = consumeMsgConcurrently.consumeMsg();


                    return null;
                }
            });

            try {
                consumer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
            }


        });


    }

    private DefaultMQPushConsumer getConsumerInstance(ConsumerProperties cp, Map<String, Object> pros) {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                cp.getConsumerGroup(),
                getAclRPCHook(this.defaultProperties.getAccessKey(), this.defaultProperties.getSecretKey()),
                new AllocateMessageQueueAveragelyByCircle(),
                cp.isEnableMsgTrace(),
                cp.getCustomizedTraceTopic());
        consumer.setNamesrvAddr(this.defaultProperties.getNameServer());
        String[] topicArr = cp.getTopics().split(";");
        Arrays.stream(topicArr).forEach(t -> {
            try {
                consumer.subscribe(t, "*");
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        });

        consumer.setInstanceName(this.defaultProperties.getNameServer().split(";")[0]);

        return consumer;

    }

    private RPCHook getAclRPCHook(String ak, String sk) {
        return new AclClientRPCHook(new SessionCredentials(ak, sk));
    }


}
